package com.nyt.report.service.impl;

import com.nyt.report.service.KafkaService;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    KafkaProducer kafkaProducer;

    @Override
    public void publishMsg(String topic, String msg) {
        ProducerRecord record = new ProducerRecord(topic, msg);
        kafkaProducer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e != null && e instanceof Exception) {
                    e.printStackTrace();
                }
                else {
                    System.out.println("消息发送成功，topic：" + metadata.topic()
                            + " partition：" + metadata.partition()
                            + " offset：" + metadata.offset()
                            + " timestamp：" + metadata.timestamp());
                }
            }
        });
    }

}
